package drds.server.datasource;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.ErrorCode;
import drds.server.handler.ConnectionHeartBeatHandler;
import drds.server.handler.Handler;
import drds.server.net.SessionContextGroupShareResources;
import drds.server.net.packet.EofPacket;
import drds.server.net.packet.ErrorPacket;
import drds.server.net.packet.FieldPacket;
import drds.server.net.packet.OkPacket;
import drds.server.net.packet.ResultSetHeaderPacket;
import drds.server.net.packet.RowDataPacket;
import drds.server.route.RouteNode;
import drds.server.server.SessionContext;
import drds.server.server.parser.ServerParse;
import drds.server.util.CurrentTimeMillis;
import drds.server.util.ResultSetUtil;
import drds.server.util.StringUtil;

/**
 * 当收回数据库连接的时候模式设置为自动提交且读写模式为写。
 * 
 * @author 曹正辉<330937205@qq.com>
 */
public class ConnectionHolder implements IConnectionHolder {
	protected static final Logger LOGGER = LoggerFactory.getLogger(ConnectionHolder.class);
	private DataSource dataSource;
	/**
	 * RDS实例管控
	 */
	private volatile String schema;
	private volatile String oldSchema;

	private byte packetId;
	private volatile boolean running = false;
	private volatile boolean borrowed;
	private long id = 0;
	private String host;
	private int port;
	private Connection connection;
	private Handler handler;
	private volatile Object attachement;

	boolean headerOutputed = false;
	private volatile boolean modifiedSQLExecuted;
	private final long startTime;
	private long lastTime;

	private SessionContextGroupShareResources sessionContextGroupShareResources;

	public SessionContextGroupShareResources getSessionContextGroupShareResources() {
		return sessionContextGroupShareResources;
	}

	public void setSessionContextGroupShareResources(SessionContextGroupShareResources sessionContextGroupShareResources) {
		this.sessionContextGroupShareResources = sessionContextGroupShareResources;
	}

	public ConnectionHolder() {
		startTime = System.currentTimeMillis();
	}

	public Connection getConnection() {
		return connection;
	}

	public void seConnection(Connection connection) {
		this.connection = connection;

	}

	@Override
	public void close(String reason) {
		try {
			connection.close();
			if (sessionContextGroupShareResources != null) {
				sessionContextGroupShareResources.removeConnectionHolder(this);
			}

		} catch (SQLException e) {
		}

	}

	public void setId(long id) {
		this.id = id;
	}

	public DataSource getDataSource() {
		return dataSource;
	}

	public void setDataSource(DataSource dataSource) {
		this.dataSource = dataSource;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public void setPort(int port) {
		this.port = port;
	}

	@Override
	public boolean isClosed() {
		try {
			return connection == null || connection.isClosed();
		} catch (SQLException e) {
			return true;
		}
	}

	@Override
	public void idleCheck() {
		long IdleTimeout = 0l;// 统一配置超时时间
		if (CurrentTimeMillis.currentTimeMillis() > lastTime + IdleTimeout) {
			close(" idle  check");
		}
	}

	public long getStartTime() {
		return startTime;
	}

	public String getHost() {
		return this.host;
	}

	public int getPort() {
		return this.port;
	}

	@Override
	public boolean isModifiedSQLExecuted() {
		return modifiedSQLExecuted;
	}

	@Override
	public boolean isSlaveDataBase() {
		return false;
	}

	@Override
	public String getSchema() {
		return this.schema;
	}

	@Override
	public void setSchema(String schema) {
		this.oldSchema = this.schema;
		this.schema = schema;
	}

	@Override
	public long getLastTime() {
		return lastTime;
	}

	@Override
	public boolean isClosedOrQuit() {
		return this.isClosed();
	}

	@Override
	public void setAttachment(Object attachment) {
		this.attachement = attachment;

	}

	@Override
	public void quit() {
		this.close("client quit");

	}

	@Override
	public void setLastTime(long lastTime) {
		this.lastTime = lastTime;

	}

	@Override
	public void release() {
		modifiedSQLExecuted = false;
		setHandler(null);
		dataSource.releaseChannel(this);
	}

	public void setRunning(boolean running) {
		this.running = running;

	}

	@Override
	public boolean setHandler(Handler handler) {
		this.handler = handler;
		return false;
	}

	@Override
	public void commit() {
		try {
			connection.commit();
			this.handler.okResponse(this, OkPacket.OK);
		} catch (SQLException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 
	 * @param node
	 * @param sessionContext
	 * @param autoCommit
	 * @throws IOException
	 */
	private void executeSql(RouteNode node, SessionContext sessionContext, boolean autoCommit) throws IOException {
		String nodeSql = node.getNodeSql();
		if (!modifiedSQLExecuted && node.isModifySQL()) {
			modifiedSQLExecuted = true;
		}

		try {
			if (!this.schema.equals(this.oldSchema)) {
				connection.setCatalog(schema);
				this.oldSchema = schema;
			}
			connection.setAutoCommit(autoCommit);
			int sqlType = node.getSqlType();
			if (sqlType == ServerParse.SELECT) {
				executeQuery(sessionContext, nodeSql);
			} else {
				executeUpdate(sessionContext, nodeSql);
			}

		} catch (SQLException e) {

			String msg = e.getMessage();
			ErrorPacket errorPacket = new ErrorPacket();
			errorPacket.packetId = ++packetId;
			errorPacket.errorNo = e.getErrorCode();
			errorPacket.errorMessage = msg.getBytes();
			this.handler.errorResponse(this, errorPacket.writeToBytes(sessionContext));
		} catch (Exception e) {
			String errorMessage = e.getMessage();
			ErrorPacket errorPacket = new ErrorPacket();
			errorPacket.packetId = ++packetId;
			errorPacket.errorNo = ErrorCode.ER_UNKNOWN_ERROR;
			errorPacket.errorMessage = errorMessage.getBytes();
			String err = null;
			if (errorPacket.errorMessage != null) {
				err = new String(errorPacket.errorMessage);
			}
			LOGGER.error("sql execute error, " + err + ", " + e);
			this.handler.errorResponse(this, errorPacket.writeToBytes(sessionContext));
		} finally {
			this.running = false;
		}

	}

	private void executeUpdate(SessionContext sessionContext, String sql) throws SQLException {
		Statement statement = null;
		try {
			statement = connection.createStatement();
			int affectedRowCount = statement.executeUpdate(sql);
			OkPacket okPacket = new OkPacket();
			okPacket.affectedRowCount = affectedRowCount;
			okPacket.insertId = 0;
			okPacket.packetId = ++packetId;
			okPacket.message = " OK!".getBytes();
			this.handler.okResponse(this, okPacket.writeToBytes(sessionContext));
		} finally {
			if (statement != null) {
				try {
					statement.close();
				} catch (SQLException e) {

				}
			}
		}
	}

	private void executeQuery(SessionContext sessionContext, String sql) throws SQLException {

		Statement statement = null;
		ResultSet resultSet = null;
		try {
			statement = connection.createStatement();
			resultSet = statement.executeQuery(sql);
			// M
			List<FieldPacket> fieldPacketList = new LinkedList<FieldPacket>();
			ResultSetUtil.resultSetToFieldPacket("", fieldPacketList, resultSet);

			ResultSetHeaderPacket resultSetHeaderPacket = new ResultSetHeaderPacket();
			resultSetHeaderPacket.fieldCount = fieldPacketList.size();
			resultSetHeaderPacket.packetId = ++packetId;

			ByteBuffer byteBuffer = sessionContext.allocate();
			byteBuffer = resultSetHeaderPacket.write(byteBuffer, sessionContext, true);
			byteBuffer.flip();
			byte[] headerData = new byte[byteBuffer.limit()];
			byteBuffer.get(headerData);
			byteBuffer.clear();

			List<byte[]> fieldBytesList = new ArrayList<byte[]>(fieldPacketList.size());
			Iterator<FieldPacket> iterator = fieldPacketList.iterator();
			while (iterator.hasNext()) {
				FieldPacket fieldPacket = iterator.next();
				fieldPacket.packetId = ++packetId;
				byteBuffer = fieldPacket.write(byteBuffer, sessionContext, false);
				byteBuffer.flip();
				byte[] field = new byte[byteBuffer.limit()];
				byteBuffer.get(field);
				byteBuffer.clear();
				fieldBytesList.add(field);
				iterator.remove();
			}
			EofPacket eofPacket = new EofPacket();
			eofPacket.packetId = ++packetId;
			byteBuffer = eofPacket.write(byteBuffer, sessionContext, false);
			byteBuffer.flip();
			byte[] eof = new byte[byteBuffer.limit()];
			byteBuffer.get(eof);
			byteBuffer.clear();
			this.handler.fieldEofResponse(this, headerData, fieldBytesList, eof);

			int colunmCount = fieldPacketList.size();
			// output row data
			while (resultSet.next()) {
				RowDataPacket rowDataPacket = new RowDataPacket(colunmCount);
				for (int i = 0; i < colunmCount; i++) {
					rowDataPacket.add(StringUtil.getBytes(resultSet.getString(i + 1)));
				}
				rowDataPacket.packetId = ++packetId;
				byteBuffer = rowDataPacket.write(byteBuffer, sessionContext, false);
				byteBuffer.flip();
				byte[] row = new byte[byteBuffer.limit()];
				byteBuffer.get(row);
				byteBuffer.clear();
				this.handler.rowResponse(this, row);
			}

			// end row
			eofPacket = new EofPacket();// -2
			eofPacket.packetId = ++packetId;
			byteBuffer = eofPacket.write(byteBuffer, sessionContext, false);
			byteBuffer.flip();
			eof = new byte[byteBuffer.limit()];
			byteBuffer.get(eof);
			sessionContext.recycle(byteBuffer);
			this.handler.rowEofResponse(this, eof);
		} finally {
			if (resultSet != null) {
				try {
					resultSet.close();
				} catch (SQLException e) {

				}
			}
			if (statement != null) {
				try {
					statement.close();
				} catch (SQLException e) {

				}
			}
		}
	}

	@Override
	public void query(final String sql) throws UnsupportedEncodingException {
		if (handler instanceof ConnectionHeartBeatHandler) {
			justForHeartbeat(sql);
		} else {
			throw new UnsupportedEncodingException("unsupported yet ");
		}
	}

	private void justForHeartbeat(String sql) {

		Statement statement = null;

		try {
			statement = connection.createStatement();
			statement.execute(sql);
			if (!isAutoCommit()) {
				// 如果在写库上，如果是事务方式的连接，需要进行手动commit
				connection.commit();
			}
			this.handler.okResponse(this, OkPacket.OK);

		} catch (Exception e) {
			String msg = e.getMessage();
			ErrorPacket error = new ErrorPacket();
			error.packetId = ++packetId;
			error.errorNo = ErrorCode.ER_UNKNOWN_ERROR;
			error.errorMessage = msg.getBytes();
			this.handler.errorResponse(this, error.writeToBytes());
		} finally {
			if (statement != null) {
				try {
					statement.close();
				} catch (SQLException e) {

				}
			}
		}
	}

	@Override
	public Object getAttachment() {
		return this.attachement;
	}

	public String getCharset() {
		return null;
	}

	@Override
	public void execute(final RouteNode node, final SessionContext sessionContext, final boolean autoCommit) throws IOException {
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				try {
					executeSql(node, sessionContext, autoCommit);
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
			}
		};

		Server.getInstance().getBusinessExecutor().execute(runnable);
	}

	@Override
	public boolean syncAndExcute() {
		return true;
	}

	@Override
	public void rollback() {
		try {
			connection.rollback();

			this.handler.okResponse(this, OkPacket.OK);
		} catch (SQLException e) {
			throw new RuntimeException(e);
		}
	}

	public boolean isRunning() {
		return this.running;
	}

	@Override
	public boolean isBorrowed() {
		return this.borrowed;
	}

	@Override
	public void setBorrowed(boolean borrowed) {
		this.borrowed = borrowed;

	}

	@Override
	public int getTransactionIsolation() {
		if (connection != null) {
			try {
				return connection.getTransactionIsolation();
			} catch (SQLException e) {
				return 0;
			}
		} else {
			return -1;
		}
	}

	@Override
	public boolean isAutoCommit() {
		if (connection == null) {
			return true;
		} else {
			try {
				return connection.getAutoCommit();
			} catch (SQLException e) {

			}
		}
		return true;
	}

	@Override
	public long getId() {
		return id;
	}

	@Override
	public void discardClose(String reason) {

	}

}
